package com.to8to.kitt.esb;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;

import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.util.AbstractMap;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.TreeMap;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

import org.apache.commons.configuration.PropertiesConfiguration;
import org.apache.commons.lang3.StringUtils;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.ChildData;
import org.apache.curator.framework.recipes.cache.NodeCache;
import org.apache.curator.framework.recipes.cache.NodeCacheListener;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.utils.EnsurePath;
import org.codehaus.jackson.JsonNode;
import org.codehaus.jackson.JsonProcessingException;
import org.codehaus.jackson.map.ObjectMapper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.base.Splitter;
import com.to8to.kitt.strategy.PriorityStrategy;
import com.to8to.kitt.strategy.RandomStrategy;
import com.to8to.kitt.strategy.RoundRobinStrategy;
import com.to8to.kitt.strategy.SelectorStrategy;
import com.to8to.kitt.utils.DateUtil;
import com.to8to.kitt.utils.LogPatterns;
import com.to8to.kitt.utils.LogUtils;
import com.to8to.kitt.utils.NettyUtils;


public class ZKServiceChannelPool implements PathChildrenCacheListener{

	public static Logger logger = LoggerFactory.getLogger(ZKServiceChannelPool.class);  
	
//	private Random random = new Random();
	
	private ObjectMapper om = new ObjectMapper();
	
	private Map<String, Map<String, Channel> > serviceMap = new HashMap<String, Map<String, Channel>> ();
	
	private PathChildrenCache servicePathCache;
	
	private Map<String, PathChildrenCache> service2childcache = new HashMap<String, PathChildrenCache>();

	private Map<String, SelectorStrategy<String> > service2strategy = new HashMap<String, SelectorStrategy<String>>();
	
	private Map<String, List<String> > service2instancelist = new HashMap<String, List<String>>();
	
	private Bootstrap bootstrap;
	
//	private String connectionDump;
	
	private ReadWriteLock lock = new ReentrantReadWriteLock();     
	
	
	public ZKServiceChannelPool(Bootstrap bootstrap)
	{
		this.bootstrap = bootstrap;
	}
	
	public void init() 
	{	
		logger.info("ZkServiceChannelPool::init()---------start!");
		
		try {
			initServicesManually();
			
			logger.info("监听服务列表: " + ZK.ZK_CONFIG.servicePath);
			
			EnsurePath ensurePath = new EnsurePath(ZK.ZK_CONFIG.servicePath);
			
			ensurePath.ensure(ZK.ZK_CLIENT.getZookeeperClient());
			
			servicePathCache = new PathChildrenCache(ZK.ZK_CLIENT, ZK.ZK_CONFIG.servicePath , true, false, ZK.ZkCuratorThreadPool);
        	
			servicePathCache.start();
			
			servicePathCache.getListenable().addListener(this);
	        
			
		} catch (Exception e) {
			logger.error(LogUtils.logStackTrace(e));
		}
		
		logger.info("ZkServiceChannelPool::init()---------finish!");
		
		
//		///channel破坏
//		BrokenRunnable r = new BrokenRunnable();
//		r.pool = this;
//		Thread t = new Thread(r);
//		t.start();
	}
	
	
	public static class BrokenRunnable implements Runnable{

		public ZKServiceChannelPool pool;
		
		@Override
		public void run() {
			
			try {
				Thread.sleep(3000);
			} catch (InterruptedException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
			
			while(true)
			{
				System.err.println("开始channel破坏行为,哈哈哈哈哈@_@ @_@ @_@");
				
				for(Map<String, Channel> map : pool.serviceMap.values())
				{
					for(Channel channel : map.values())
					{
						System.err.println("close channel: " + channel.toString());
						
						channel.close();
						
//						try {
//							Thread.sleep(1000);
//						} catch (InterruptedException e) {
//							// TODO Auto-generated catch block
//							e.printStackTrace();
//						}
					}
				}
				
				
				try {
					Thread.sleep(20000);
				} catch (InterruptedException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}
			}
		}
	}
	
	
	private void initServicesManually() throws Exception
	{
		long t1 = System.currentTimeMillis();
		
		List<String> services = ZK.ZK_CLIENT.getChildren().forPath(ZK.ZK_CONFIG.servicePath);
		for(String service : services)
		{
			String servicePath = ZK.ZK_CONFIG.servicePath + "/" + service;
			
			byte[] serviceData = ZK.ZK_CLIENT.getData().forPath(servicePath);
			
			PathChildrenCacheEvent serviceEvent = new PathChildrenCacheEvent(PathChildrenCacheEvent.Type.CHILD_ADDED
					, new ChildData(servicePath, null, serviceData));
			
			logger.info("手动触发事件:{}",  servicePath);
			
			this.childEvent(ZK.ZK_CLIENT, serviceEvent);
			
			List<String> instances = ZK.ZK_CLIENT.getChildren().forPath(servicePath);
			for(String instance : instances)
			{
				String instancePath = servicePath + "/" + instance;
				byte[] instanceData = ZK.ZK_CLIENT.getData().forPath(instancePath);
				PathChildrenCacheEvent instanceEvent = new PathChildrenCacheEvent(PathChildrenCacheEvent.Type.CHILD_ADDED
						, new ChildData(instancePath, null, instanceData));
				
				logger.info("手动触发事件:{}",  instancePath);
				
				this.childEvent(ZK.ZK_CLIENT, instanceEvent);
			}
		}
		
		long t2 = System.currentTimeMillis();
		
		logger.info("###########################################初始化加载成功，耗时:{} ms", (t2-t1));
	}
	
	public String dumpServiceConnectionsMT() throws Exception
	{
		
		
		Map<String, Map<String, Map<String, String>>> bigmap = new TreeMap<String, Map<String, Map<String, String>>>();
		
		Set<String> brokenServiceSet = new HashSet<String>();
		String newDump = null;
		
		lock.readLock().lock();
		
		try{

			
			for(Entry<String, Map<String, Channel>> entry : serviceMap.entrySet())
			{
				String service = entry.getKey();
				
				boolean hasBrokenChannel = false;
				
				Map<String, Map<String, String>> m1 = new TreeMap<String, Map<String, String>>();
				
				bigmap.put(service, m1);
				
				Map<String, Channel> map = entry.getValue();
				
				for(Entry<String, Channel> subentry : map.entrySet())
				{
					Map<String, String> m2 = new TreeMap<String, String>();
					
					String instance = subentry.getKey();
					
					m1.put(instance, m2);
					
					Channel channel = subentry.getValue();
					
//					boolean isOpen = channel.isOpen();
//					boolean isActive = channel.isActive();
//					boolean isRegistered = channel.isRegistered();
//					boolean isWritable = channel.isWritable();
					
					m2.put("channel", channel.toString());
					
					boolean isOK = NettyUtils.checkChannelHealth(channel);
					
					m2.put("isOK",  isOK + "");
					
					if(!isOK)
					{
						hasBrokenChannel = true;
					}
				}
				
				if(hasBrokenChannel)
					brokenServiceSet.add(service);
				
				newDump = om.writeValueAsString(bigmap);
			}
		}
		catch(Exception ee)
		{
			logger.error(LogUtils.logStackTrace(ee));
		}
		finally
		{
			lock.readLock().unlock();
		}
		
		for(String service : brokenServiceSet)
			reload(service);
		
		return newDump;
	}
	
	public Entry<String, Channel> getChannelMT(String service, String address)
	{
		lock.readLock().lock();
		
		Entry<String, Channel> result = null;
		
		try
		{
			Map<String, Channel> instance2channelpool = serviceMap.get(service);
			

			
			if(instance2channelpool != null)
			{
				for(Entry<String, Channel> pool : instance2channelpool.entrySet())
				{
					if(pool.getValue().remoteAddress().toString().contains(address))
					{
						result = pool;
						
						return result;
					}
				}
			}
		}
		catch(Exception ee)
		{
			logger.error(LogUtils.logStackTrace(ee));
		}
		finally
		{
			lock.readLock().unlock();
		}
		
		return result;
	}
	
	public Entry<String, Channel> getChannelMT(String service) 
	{
		RoutineResult result = route(service);
		
		if(result == null || !NettyUtils.checkChannelHealth(result.channel))
		{
			if(result != null)
			{
				logger.error("{} 路由到下游链路，但是链路损坏！尝试关闭并删除坏链: 服务:{} 实例:{} 链路:{}", LogPatterns.ZK_CONN_TAG
						, result.service, result.instance, result.channel);				
				close(result);
			}
			else 
				logger.error("{} 找不到下游的链路, 服务: {}", LogPatterns.ZK_CONN_TAG, service);	

			reload(service);
			
			logger.info("再路由一次试试: {}", service);
			
			result = route(service);
			
			if(result != null)
			{
				logger.info("终于路由成功: 实例:{}, 连接:{}", result.instance, result.channel);
			}
			else 
				logger.error("还是路由失败！");
		}
		if(result != null)
			return new AbstractMap.SimpleEntry<String, Channel>(result.instance, result.channel);
		return null;
	}
	
	public static class RoutineResult 
	{
		public String service;
		public String instance;
		public Channel channel;
		public Map<String, Channel> instanceMap;
	}
	
	/**
	 * 
	 * 服务的实例路由
	 * @param service
	 * @return
	 */
	private RoutineResult route(String service) 
	{
		RoutineResult result = null;
		
		lock.readLock().lock();
		
		try
		{
			Map<String, Channel> instanceMap = serviceMap.get(service);
			
			if(instanceMap != null && instanceMap.size() > 0)
			{
				SelectorStrategy<String> strategy = service2strategy.get(service);
				
				if(strategy == null)
					strategy = new RandomStrategy<String>();
				
				List<String> instances = service2instancelist.get(service);
				
				if(instances == null)
				{
					instances = new ArrayList<String>();
					instances.addAll(instanceMap.keySet());
				}
				
				logger.debug("服务{}的路由策略为:{}", service, strategy.getClass().getSimpleName());
				
				List<String> validInstances = new ArrayList<String>();
				
				for(String instance : instances)
				{
					if(instanceMap.containsKey(instance))
						validInstances.add(instance);
				}
				
				String instance = strategy.select(validInstances);
				
				if(instance != null)
				{
					Channel channel = instanceMap.get(instance);
					result = new RoutineResult();
					result.service = service;
					result.instanceMap = instanceMap;
					result.instance = instance;
					result.channel = channel;
				}
				else 
				{
					logger.error("服务:{}路由到的服务实例为空! \ninstances:{} \ninstanceMap:{} \nvalidInstances:{} \nselected instance:{}"
							, service, instances, instanceMap, validInstances, instance);
				}
			}
			else 
			{
				logger.error("{} 服务{}的instanceMap不存在，或者为空：{}, serviceMap:{}", LogPatterns.ZK_CONN_TAG , service, instanceMap, serviceMap);
			}
		}
		catch(Exception e)
		{
			
		}
		finally
		{
			lock.readLock().unlock();
		}
		
		return result;
	}
	
	private void reload(String service)
	{
		logger.info("{} 重新加载ZK的服务:{}，并建立连接: ", LogPatterns.ZK_CONN_TAG, service);
		
		String servicePath = ZK.ZK_CONFIG.servicePath + "/" + service;
		
		List<String> children = null; 
		
		try {
			children = ZK.ZK_CLIENT.getChildren().forPath(servicePath);
		} catch (Exception e) {
			e.printStackTrace();
			logger.error(LogUtils.logStackTrace(e));
			logger.error("{} 从zk上查找到服务的实例列表  失败啦！！！ 啥都干不了啦！！！" + servicePath, LogPatterns.ZK_CONN_TAG);
			return ;
		}
		
		if(children == null || children.isEmpty())
		{
			logger.error("{} 从zk上查找到服务的实例列表为空: {}, 啥都干不了啦！", LogPatterns.ZK_CONN_TAG, servicePath);
			return ;
		}
		
		Map<String, Channel> instanceMap = serviceMap.get(service);
		
		if(instanceMap != null)
		{
			for(Entry<String, Channel> entry : instanceMap.entrySet())
			{
				RoutineResult rr = new RoutineResult();
				rr.channel = entry.getValue();
				rr.instance = entry.getKey();
				rr.service = service;
				rr.instanceMap = instanceMap;
				close(rr);
			}
		}
		
		for(String instance : children)
		{
			logger.info("{} 读取zk的实例数据: service: {}, instance:{}", LogPatterns.ZK_CONN_TAG, service, instance);
			
			String instancePath = servicePath + "/" + instance;
			
			byte[] data = null;
			
			try {
				data = ZK.ZK_CLIENT.getData().forPath(instancePath);
			} catch (Exception e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
			
			if(data == null)
			{
				logger.error("{} 读取数据失败！！ service:{}, instance:{}", LogPatterns.ZK_CONN_TAG, service, instance);
				continue;
			}
        	
        	JsonNode jsonNode = null;
        	
			try {
				jsonNode = om.readTree(data);
			} catch (JsonProcessingException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			} catch (IOException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
			
			if(jsonNode != null)
			{
	        	String host = jsonNode.get("address").asText();
	        	
	        	int port = jsonNode.get("port").asInt();
	        	
	        	connect(host, port, service, instance, instanceMap);
			}
		}
	}

	
	private void ensureConnect(String service, String instance, String host, int port)
	{
		if(!isValidService(service, instance, host, port))
		{
			logger.warn("非法的服务:{} 实例:{} 地址:{}", service, instance, host + ":" + port);
			return;
		}
		
		lock.writeLock().lock();
		
		try
		{
			Map<String, Channel> instanceMap = serviceMap.get(service);
			
			if (instanceMap == null)
			{
				instanceMap = new HashMap<String, Channel>();
				
				serviceMap.put(service, instanceMap);
			}
			
			Channel channel = instanceMap.get(instance);
			
			if (channel == null || !NettyUtils.checkChannelHealth(channel))
			{
	    		logger.info("{} [+++++++] 连接服务[{}]的实例[{}] 地址 : [{}:{}]", LogPatterns.ZK_CONN_TAG, service, instance, host, port);
	    		
	    		connect( host, port, service, instance, instanceMap);
			}
			else 
				logger.info("channel is health, ignore, 服务:{} 实例:{} 链路:{}", service, instance, channel);
		}
		catch(Exception ee)
		{
			logger.error(LogUtils.logStackTrace(ee));
		}
		finally
		{
			lock.writeLock().unlock();
		}
	}
	
	
	private void ensureClose(String service, String instance)
	{
		logger.info("{} [-------] 尝试关闭与服务[{}]的实例[{}]的连接", LogPatterns.ZK_CONN_TAG, service, instance);
		
		Map<String, Channel> instanceMap = serviceMap.get(service);
		
		if (instanceMap != null)
		{
			Channel channel = instanceMap.get(instance);
			
			if(channel != null)
			{
				RoutineResult rr = new RoutineResult();
				rr.service = service;
				rr.instance = instance;
				rr.channel = channel;
				rr.instanceMap = instanceMap;
				close(rr);
			}
			else 
			{
				logger.info("{} 服务[{}]的实例[{}]的连接信息本来就不存在，忽略之", LogPatterns.ZK_CONN_TAG, service, instance);
				instanceMap.remove(instance);
			}
		}
		else 
		{
			logger.info("{} 服务{}的连接信息本来就不存在，忽略之", LogPatterns.ZK_CONN_TAG, service);
		}
	}
	
	public void ensureCloseMT(String service, String instance)
	{
		lock.writeLock().lock();
		
		try
		{
			ensureClose(service, instance);
		}
		catch(Exception ee)
		{
			logger.error(LogUtils.logStackTrace(ee));
		}
		finally
		{
			lock.writeLock().unlock();
		}
	}
	
	private void close(RoutineResult result)
	{
		logger.info("{} 尝试关闭并删除坏链, 服务:{} 实例:{} 链路:{}", LogPatterns.ZK_CONN_TAG
				, result.service, result.instance, result.channel);
		
		ChannelFuture cf = result.channel.close();
		
		cf.addListener(new ChannelFutureListener(){
			@Override
			public void operationComplete(ChannelFuture future)
					throws Exception {
				
				if(future.isSuccess())
					logger.info("{} 断开连接成功! 服务:{} 实例:{} 链路:{}", LogPatterns.ZK_CONN_TAG, result.service, result.instance, result.channel);
				else 
					logger.error("{} 断开连接失败! 服务:{} 实例:{} 链路:{}", LogPatterns.ZK_CONN_TAG, result.service, result.instance, result.channel);
				
				future.removeListener(this);
			}
		});
		
		result.instanceMap.remove(result.instance);
	}
	
	private Channel connect(String host, int port, String service, String instance, Map<String, Channel> instanceMap)
	{
		if(!isValidService(service, instance, host, port))
		{
			logger.warn("非法的服务:{} 实例:{} 地址:{}", service, instance, host + ":" + port);
			return null;
		}
		
		logger.info("{} 尝试连接    服务:{} 实例:{} 链路:{}", LogPatterns.ZK_CONN_TAG, service, instance, host + ":" + port);
		
		boolean sync = true;
		
		ChannelFutureListener l = new ChannelFutureListener(){
			@Override
			public void operationComplete(ChannelFuture future)
					throws Exception {
				Channel channel = future.channel();
				if(future.isSuccess())
				{
					logger.info("{} 连接成功!! 服务:{} 实例:{} 链路:{}", LogPatterns.ZK_CONN_TAG, service, instance, channel);
					instanceMap.put(instance, channel);
				}
				else  
				{
					logger.error("{} 连接失败!! 服务:{} 实例:{} 链路:{}", LogPatterns.ZK_CONN_TAG, service, instance, channel);
				}
				
				future.removeListener(this);
			}
		};
		
		ChannelFuture f = null;
		if(sync)
		{
			try {
				f = bootstrap.connect(host, port).sync();
				l.operationComplete(f);
			} catch (Exception e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
				return null;
			}
		}
		else 
		{
			f = bootstrap.connect(host, port);
			f.addListener(l);
		}
		
		return f != null? f.channel() : null;
	}
	
	@Override
	public void childEvent(CuratorFramework client, PathChildrenCacheEvent event)
			throws Exception {

		try{
			if(event == null)
			{
				logger.error("{} event is null!", LogPatterns.ZK_CONN_TAG);
				return ;
			}
			
			ChildData cdata = event.getData();
			if(cdata == null)
			{
				logger.error("{} cdata is null!", LogPatterns.ZK_CONN_TAG);
				return ;
			}
			
			String path = cdata.getPath();
			
			System.err.println(DateUtil.format(new Date()) + " ZK event path:" + path + ", type:" + event.getType());
			
			logger.info("{} ZK event path:" + path + ", type:" + event.getType(), LogPatterns.ZK_CONN_TAG);
			
			String remain = StringUtils.substringAfter(path, ZK.ZK_CONFIG.servicePath + "/");
			
			if(!remain.contains("/"))
			{
				String servicePath = path;
				
				switch ( event.getType() )
		        {
		            case CHILD_ADDED:
		            {
		            	PathChildrenCache childcache = service2childcache.get(servicePath);
		            	
		            	if(childcache != null)
		            	{
		            		logger.info("{} 服务已经监听:{}, 忽略之", LogPatterns.ZK_CONN_TAG, servicePath);
		            		break;
		            	}
		            	
		            	logger.info("{} ==========>监听总线服务:{}", LogPatterns.ZK_CONN_TAG, servicePath);
		            	
		            	childcache = new PathChildrenCache(client, servicePath , true, false, ZK.ZkCuratorThreadPool);
			        	
						childcache.start();
						
				        childcache.getListenable().addListener(this);
				        
				        NodeCache nodecahe = new NodeCache(client, servicePath);
				        
				        String service = StringUtils.substringAfterLast(servicePath, "/");
				        
				        nodecahe.start();
				        
				        nodecahe.getListenable().addListener(new NodeCacheListener(){
				        	
							@Override
							public void nodeChanged() throws Exception {
								
								byte[] data = client.getData().forPath(servicePath);
								
								String txt = new String(data);
								
								logger.info("{} 加载服务{}的路由策略:\n----------------------------------------------------\n"
										+ "{}"
										+ "\n----------------------------------------------------", LogPatterns.ZK_CONN_TAG, servicePath, txt);
								
								if(!StringUtils.isEmpty(txt.trim()))
								{
									try {
										
										PropertiesConfiguration pc = new PropertiesConfiguration();
										
										pc.load(new ByteArrayInputStream(data));
										
										String policy = pc.getString("policy");
										
										String instances = pc.getString("instances");
										
										SelectorStrategy<String> ss = null;
										
										if("priority".equals(policy))
											ss = new PriorityStrategy<String>();
										else if("random".equals(policy))
											ss = new RandomStrategy<String>();
										else if("roundrobin".equals(policy))
											ss = new RoundRobinStrategy<String>();
										
										if(ss != null)
											service2strategy.put(service, ss);
										
										if(!StringUtils.isEmpty(instances))
										{
											List<String> ints = Splitter.on(";").splitToList(instances);
											if(ints != null)
												service2instancelist.put(service, ints);
										}
									} catch (Exception e) {
										logger.error(LogUtils.logStackTrace(e));
									}
								}
								else 
								{
									service2strategy.remove(service);
									service2instancelist.remove(service);
								}
							}});
				        
				        service2childcache.put(servicePath, childcache);
		            }
		            default:
		        }
			}
			else 
			{
				String instance = StringUtils.substringAfterLast(path, "/");
		    	
		    	remain = StringUtils.substringBeforeLast(path, "/");
		    	
		    	String service = StringUtils.substringAfterLast(remain, "/");
		    	
				switch ( event.getType() )
		        {
		            case CHILD_ADDED:
		            {
		            	byte[] data2 = cdata.getData();
		            	
		            	JsonNode jsonNode = om.readTree(data2);
		            	
		            	String host = jsonNode.get("address").asText();
		            	
		            	int port = jsonNode.get("port").asInt();
		            	
		            	ensureConnect(service, instance, host, port);
		            	
		                break;
		            }

		            case CHILD_UPDATED:
		            {
		                break;
		            }

		            case CHILD_REMOVED:
		            {
		            	ensureCloseMT(service, instance);
		                
		                break;
		            }
		            default:
		        }
			}
		}
		catch(Exception e)
		{
			logger.error(LogUtils.logStackTrace(e));
			e.printStackTrace();
		}
		
	}

	public String getServiceInstancePayload(String service, String instance) {
		
		String instancePath = ZK.ZK_CONFIG.configPath + "/" + service + "/" + instance;
		
		try {
			byte[] data = ZK.ZK_CLIENT.getData().forPath(instancePath);
			
        	JsonNode jsonNode = om.readTree(data);
        	
        	String payload = jsonNode.get("payload").asText();
        	
        	return payload;
        	
		} catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		return null;
	}
//
//	public void setConnectionDump(String connectionDump) {
//		this.connectionDump = connectionDump;
//	}

	public static boolean isValidService(String service, String instance, String host, int port)
	{
		return !StringUtils.isEmpty(host) && port > 0 && !BusServer.class.getName().equals(service);
	}

}
